package com.webflux.config;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * web socket Session 类
 *
 * @author 晨曦
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Slf4j
public class WebSocketWrap {

    public static final Map<String, WebSocketWrap> SENDER = new ConcurrentHashMap<>();

    private String id;
    /**
     * 待操作websocket连接会话
     **/
    private WebSocketSession session;
    /**
     * websocket响应堆栈操作API
     **/
    private FluxSink<WebSocketMessage> sink;

    /**
     * 关闭用户Session连接
     *
     * @param id
     */
    public static Mono<Void> closeSession(String id) {
        WebSocketWrap wrap = SENDER.get(id);
        if (!ObjectUtils.isEmpty(wrap)) {
            SENDER.remove(wrap.getId());
            log.info("关闭用户->{}Session", id);
            return  wrap.getSession().close()
                    .log()
                    .doOnError(throwable -> log.error("webSocket发生异常：" + throwable));
        }
        return wrap.session.close(new CloseStatus(1016, "连接未通过校验,即将关闭连接"));
    }

    /**
     * 发送广播消息
     *
     * @param obj 消息对象，会被转为JSON
     * @since jdk11
     */
    public static void broadcastText(Object obj) {
        SENDER.values().forEach(wrap -> wrap.sendText(obj));
    }

    /**
     * 发送文本
     *
     * @param obj
     */
    public void sendText(Object obj) {
        sink.next(session.textMessage(JSON.toJSONString(obj)));
    }


    static {
        purge();
    }

    /**
     * 清理不可用的SESSION
     *
     * @return void
     */
    public static void purge() {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            new ArrayList<>(SENDER.values()).forEach(wrap -> {
                if (!wrap.getSession().isOpen()) {
                    log.warn(String.format("用户ID: [%s] 的session: [%s] 已经关闭，将被清理", wrap.getId(), wrap.getSession().getId()));
                    SENDER.remove(wrap.getId());
                    wrap.getSession().close();
                }
            });
        }, 30, 30, TimeUnit.SECONDS);
    }
}
